package client.kafka.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.apache.log4j.Logger;

import client.kafka.IKafkaClient;
import client.kafka.KafkaClientController;
import client.kafka.KafkaTreeNode;
import client.kafka.impl.DO.KafkaConsumerGroupListTreeNode;
import client.kafka.impl.DO.KafkaConsumerGroupTreeNode;
import client.kafka.impl.DO.KafkaRootTreeNode;
import client.kafka.impl.DO.KafkaTopicListTreeNode;
import client.kafka.impl.DO.KafkaTopicTreeNode;
import client.zookeeper.ZookeeperTreeNode;
import core.AlertUtil;
import core.FormDialog;
import core.JavaProcessBuilder;
import javafx.scene.control.ContextMenu;
import javafx.scene.control.MenuItem;
import javafx.scene.control.TextArea;
import javafx.scene.control.TextField;
import javafx.scene.control.TreeItem;
import javafx.scene.control.TreeView;
import kafka.admin.AclCommand;
import kafka.admin.BrokerApiVersionsCommand;
import kafka.admin.ConfigCommand;
import kafka.admin.ConsumerGroupCommand;
import kafka.admin.DelegationTokenCommand;
import kafka.admin.DeleteRecordsCommand;
import kafka.admin.LogDirsCommand;
import kafka.admin.PreferredReplicaLeaderElectionCommand;
import kafka.admin.ReassignPartitionsCommand;
import kafka.admin.TopicCommand;
import kafka.tools.ConsoleConsumer;
import kafka.tools.ConsoleProducer;
import kafka.tools.ConsumerPerformance.ConsumerPerfConfig;
import kafka.tools.DumpLogSegments;
import kafka.tools.MirrorMaker;
import kafka.tools.ReplicaVerificationTool;
import kafka.tools.StreamsResetter;

/**
 * 单台Kafka连接客户端
 * @author self
 *
 */
public class KafkaClientImpl implements IKafkaClient {

	/**
	 * 命令行支持的命令
	 */
	private static final Map<String,Class<?>> KAFKA_COMMAND = new HashMap<String,Class<?>>();
	static {
		KAFKA_COMMAND.put("acls", AclCommand.class);
		KAFKA_COMMAND.put("broker-api-versions", BrokerApiVersionsCommand.class);
		KAFKA_COMMAND.put("configs", ConfigCommand.class);
		KAFKA_COMMAND.put("console-consumer", ConsoleConsumer.class);
		KAFKA_COMMAND.put("console-producer", ConsoleProducer.class);
		KAFKA_COMMAND.put("consumer-groups", ConsumerGroupCommand.class);
		KAFKA_COMMAND.put("consumer-perf-test", ConsumerPerfConfig.class);
		KAFKA_COMMAND.put("delegation-tokens", DelegationTokenCommand.class);
		KAFKA_COMMAND.put("delete-records", DeleteRecordsCommand.class);
		KAFKA_COMMAND.put("dump-log", DumpLogSegments.class);
		KAFKA_COMMAND.put("log-dirs", LogDirsCommand.class);
		KAFKA_COMMAND.put("mirror-maker", MirrorMaker.class);
		KAFKA_COMMAND.put("preferred-replica-election", PreferredReplicaLeaderElectionCommand.class);
		KAFKA_COMMAND.put("reassign-partitions", ReassignPartitionsCommand.class);
		KAFKA_COMMAND.put("replica-verification", ReplicaVerificationTool.class);
		KAFKA_COMMAND.put("streams-application-reset", StreamsResetter.class);
		KAFKA_COMMAND.put("topics", TopicCommand.class);
	}
	
	//客户端名称
	private static final String CLIENT_NAME = "Kafka客户端";
	
	//应用日志对象
	private Logger appLogger;
	
	//界面控制器组件对象
	private KafkaClientController controller;
	
	//管理客户端
	private AdminClient adminClient;
	
	//连接信息
	private String connStr;
	
	@Override
	public String getName() {
		return CLIENT_NAME;
	}

	@Override
	public void init(KafkaClientController controller) {
		this.controller = controller;
		this.appLogger = controller.getAppLogger();
		
		appLogger.info("初始化客户端组件！");
		createConnection();
	}
	
	/**
	 * 创建连接
	 * @return
	 */
	private boolean createConnection() {
		appLogger.info("创建Socket连接！");
		TextArea serverTextArea = new TextArea("127.0.0.1:9092");
		FormDialog newConnForm = new FormDialog("新建链接"
				,new FormDialog.Item("主机：", serverTextArea));
		Optional<Boolean> result = newConnForm.showAndWait();
		if(result.isPresent()) {
			String servers = serverTextArea.getText();
			appLogger.info("Socket连接参数 -> "+servers);
			this.connStr = servers;
			Properties kafkaProp = new Properties();
			kafkaProp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connStr);
			adminClient = KafkaAdminClient.create(kafkaProp);
			this.controller.getCmdTextArea().setText("已连接！\n");
			appLogger.info("已连接到Kafka服务器。");
			initTreeView();
		}
		return false;
	}

	/**
	 * 初始化TreeView树
	 */
	private void initTreeView() {
		appLogger.info("初始化Kafka树结构。");
		TreeView<KafkaTreeNode> kafkaTreeView = controller.getKafkaTreeView();
		appLogger.info("初始化树节点！");
		TreeItem<KafkaTreeNode> root = this.getRootTreeItem();
		kafkaTreeView.setRoot(root);
		
		appLogger.info("添加节点选择监听事件！");
		//节点选择事件处理
		kafkaTreeView.getSelectionModel().selectedItemProperty().addListener((observable,oldValue,newValue)->{
			appLogger.info("Kafka树节点被点击 -> oldValue:"+oldValue+",newValue:"+newValue);
			TreeItem<KafkaTreeNode> currentSelectedItem = (TreeItem<KafkaTreeNode>) newValue;
			if(currentSelectedItem==null) return;
			
			//设置节点信息
			KafkaTreeNode selectedNode = currentSelectedItem.getValue();
			String itemInfo = this.getItemInfo(selectedNode);
			controller.getInfoTextArea().setText(itemInfo);
			
			if(!selectedNode.isLeaf() && !selectedNode.isLoaded()) {
				currentSelectedItem.getChildren().clear();
				List<TreeItem<KafkaTreeNode>> childNodeList = this.getChildren(selectedNode);
				currentSelectedItem.getChildren().addAll(childNodeList);
				currentSelectedItem.setExpanded(true);
			}
			
			selectedNode.setLoaded(true);
		});
		
		appLogger.info("设置树节点右键菜单！");
		//设置TreeView树邮件菜单
		MenuItem refreshItem = new MenuItem("刷新"); 
		refreshItem.setOnAction((event)->{
			appLogger.info("执行[刷新]动作！");
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			
			//刷新信息
			controller.getInfoTextArea().setText(this.getItemInfo(kafkaTreeNode));
			
			//刷新子节点
			if(!kafkaTreeNode.isLeaf()) {
				selectedItem.getChildren().clear();
				selectedItem.getChildren().addAll(this.getChildren(kafkaTreeNode));
				selectedItem.setExpanded(true);
			}
		});
		
		MenuItem disconnectItem = new MenuItem("断开连接");
		disconnectItem.setOnAction((event)->{
			appLogger.info("执行[断开连接]动作！");
			try {
				this.close();
				controller.getKafkaTreeView().setRoot(null);
				controller.getKafkaTreeView().setContextMenu(null);
				controller.getInfoTextArea().clear();
				controller.getCmdTextArea().setText("连接已断开！\n");
			} catch (IOException e) {
				AlertUtil.exception(e);
			}
		});
		
		MenuItem newTopicItem = new MenuItem("新建主题");
		newTopicItem.setOnAction((event)->{
			appLogger.info("执行[新建主题]动作！");
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			TextField nameTextField = new TextField();
			TextField partitionTextField = new TextField("1");
			TextField replicationTextField = new TextField("1");
			FormDialog newTopicDialog = new FormDialog("新建主题", 
					new FormDialog.Item("主题名称", nameTextField),
					new FormDialog.Item("分区数", partitionTextField),
					new FormDialog.Item("拷贝数", replicationTextField));
			Optional<Boolean> result = newTopicDialog.showAndWait();
			if(result.isPresent()) {
				String topicName = nameTextField.getText();
				int numPartitions = Integer.valueOf(partitionTextField.getText());
				short replicationFactor = Short.valueOf(replicationTextField.getText());
				NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
				adminClient.createTopics(Arrays.asList(newTopic));
				
				//刷新子节点
				selectedItem.getChildren().clear();
				List<TreeItem<KafkaTreeNode>> childNodeList = this.getChildren(kafkaTreeNode);
				selectedItem.getChildren().addAll(childNodeList);
				selectedItem.setExpanded(true);
			}
		});
		
		MenuItem delTopicItem = new MenuItem("删除主题");
		delTopicItem.setOnAction((event)->{
			appLogger.info("执行[新建主题]动作！");
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			if(kafkaTreeNode instanceof KafkaTopicTreeNode) {
				String topic = ((KafkaTopicTreeNode) kafkaTreeNode).getTopic();
				adminClient.deleteTopics(Arrays.asList(topic));
				selectedItem.getParent().getChildren().clear(); //清理当前层级节点，重新加载
			}
		});
				
		kafkaTreeView.setContextMenu(new ContextMenu(disconnectItem)); 
		kafkaTreeView.setOnContextMenuRequested((event)->{
			kafkaTreeView.getContextMenu().getItems().clear();
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			kafkaTreeView.getContextMenu().getItems().add(refreshItem);
			if(kafkaTreeNode instanceof KafkaRootTreeNode) {
				kafkaTreeView.getContextMenu().getItems().addAll(newTopicItem,disconnectItem);
			}else if(kafkaTreeNode instanceof KafkaTopicTreeNode) {
				kafkaTreeView.getContextMenu().getItems().add(delTopicItem);
			}
		});
		
		appLogger.info("Kafka树结构初始化完成！");
	}
	
	/**
	 * 获取根节点
	 * @return　根节点对象
	 */
	private TreeItem<KafkaTreeNode> getRootTreeItem() {
		appLogger.info("获取Kafka树根节点！");
		KafkaRootTreeNode rootNode = new KafkaRootTreeNode(this.connStr);
		TreeItem<KafkaTreeNode> rootItem = new TreeItem<KafkaTreeNode>(rootNode);
		rootItem.setExpanded(true);
		rootItem.getChildren().addAll(this.getChildren(rootNode));
		return rootItem;
	}
	
	/**
	 * 获取子节点集合
	 * @param selectedNode　当前选中节点
	 * @return　子节点集合
	 */
	private List<TreeItem<KafkaTreeNode>> getChildren(KafkaTreeNode selectedNode) {
		appLogger.info("获取Kafka树子节点！当前节点："+selectedNode);
		if(selectedNode instanceof KafkaRootTreeNode) {
			List<TreeItem<KafkaTreeNode>> childrenList = new ArrayList<TreeItem<KafkaTreeNode>>();
			childrenList.add(new TreeItem<KafkaTreeNode>(new KafkaTopicListTreeNode())); //主题节点
			childrenList.add(new TreeItem<KafkaTreeNode>(new KafkaConsumerGroupListTreeNode())); //消费者节点
			return childrenList;
		}else if(selectedNode instanceof KafkaTopicListTreeNode) {
			List<TreeItem<KafkaTreeNode>> childrenList = new ArrayList<TreeItem<KafkaTreeNode>>();
			try {
				KafkaFuture<Collection<TopicListing>> topics = adminClient.listTopics().listings();
				for(TopicListing topic:topics.get()) {
					KafkaTopicTreeNode topicTreeNode = new KafkaTopicTreeNode(topic.name());
					TreeItem<KafkaTreeNode> topicItem = new TreeItem<KafkaTreeNode>(topicTreeNode);
					childrenList.add(topicItem);
				}
			} catch (InterruptedException | ExecutionException e) {
				AlertUtil.exception(e);
			}
			return childrenList;
		}else if(selectedNode instanceof KafkaConsumerGroupListTreeNode) {
			List<TreeItem<KafkaTreeNode>> childrenList = new ArrayList<TreeItem<KafkaTreeNode>>();
			try {
				KafkaFuture<Collection<ConsumerGroupListing>> consumerGroups = adminClient.listConsumerGroups().all();
				for(ConsumerGroupListing consumerGroup:consumerGroups.get()) {
					KafkaConsumerGroupTreeNode consumerGroupTreeNode = new KafkaConsumerGroupTreeNode(consumerGroup.groupId());
					TreeItem<KafkaTreeNode> consumerGroupItem = new TreeItem<KafkaTreeNode>(consumerGroupTreeNode);
					childrenList.add(consumerGroupItem);
				}
			} catch (InterruptedException | ExecutionException e) {
				AlertUtil.exception(e);
			}
			return childrenList;
		}
		return null;
	}

	/**
	 * 获取节点信息
	 * @param selectedNode　当前选中节点
	 * @return　节点信息
	 */
	private String getItemInfo(KafkaTreeNode selectedNode) {
		appLogger.info("获取Kafka树节点详情信息。当前节点："+selectedNode);
		if(selectedNode instanceof KafkaRootTreeNode) {
			StringBuilder builder = new StringBuilder();
			DescribeClusterResult result = adminClient.describeCluster();
			try {
				String clusterId = result.clusterId().get();
				String controller = String.valueOf(result.controller().get());
				String nodes = String.valueOf(result.nodes().get());
				String authorizedOperations = String.valueOf(result.authorizedOperations().get());
				builder.append("clusterId:").append(clusterId).append("\n");
				builder.append("controller:").append(controller).append("\n");
				builder.append("nodes:").append(nodes).append("\n");
				builder.append("authorizedOperations:").append(authorizedOperations).append("\n");
			} catch (InterruptedException | ExecutionException e) {
				AlertUtil.exception(e);
			}
			return builder.toString();
		}else if(selectedNode instanceof KafkaTopicTreeNode) {
			String topic = ((KafkaTopicTreeNode) selectedNode).getTopic();
			KafkaFuture<TopicDescription> result = adminClient.describeTopics(Arrays.asList(topic)).values().get(topic);
			try {
				TopicDescription topicInfo = result.get();
				return topicInfo.toString();
			} catch (InterruptedException | ExecutionException e) {
				AlertUtil.exception(e);
			}
		}else if(selectedNode instanceof KafkaConsumerGroupTreeNode) {
			String groupId = ((KafkaConsumerGroupTreeNode) selectedNode).getGroupId();
			KafkaFuture<ConsumerGroupDescription> result = adminClient.describeConsumerGroups(Arrays.asList(groupId)).describedGroups().get(groupId);
			try {
				ConsumerGroupDescription consumerGroupInfo = result.get();
				return consumerGroupInfo.toString();
			} catch (InterruptedException | ExecutionException e) {
				AlertUtil.exception(e);
			}
		}
		return null;
	}
	
	@Override
	public String exeCmd(String cmd) {
		if(cmd==null||"".equals(cmd.trim())) {
			return null;
		}
		appLogger.info("执行命令："+cmd);
		//命令如果没有换行符则补充一个
		if(!cmd.endsWith("\n")) {
			cmd+="\n";
		}
		
		//如果没有设定连接的kafka服务，添加当前连接
		if(!cmd.contains("--bootstrap-server") && !cmd.contains("--zookeeper")) {
			cmd+=" --bootstrap-server "+this.connStr;
		}
		
		//分割出命令行参数
		String[] cmdAndArgs = cmd.trim().split("\\s+");
		String command = cmdAndArgs[0];
		String[] args = Arrays.copyOfRange(cmdAndArgs, 1, cmdAndArgs.length);
		StringBuilder resultBuilder = new StringBuilder();
		if(KAFKA_COMMAND.containsKey(command)) {
			try {
				ProcessBuilder processBuilder = JavaProcessBuilder.build(KAFKA_COMMAND.get(command), args);
				processBuilder.redirectErrorStream(true); //合并输出流和错误流
				Process process = processBuilder.start();
				try(BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))){
					String line;
					while((line = reader.readLine())!=null) {
						resultBuilder.append(line+"\n");
					}
				}
				process.waitFor();
			} catch (IOException | InterruptedException e) {
				AlertUtil.exception(e);
			}
		}else {
			resultBuilder.append("命令【"+command+"】不被支持！\n\n");
			resultBuilder.append("以下是目前支持的命令列表：\n");
			for(String key:KAFKA_COMMAND.keySet()) {
				resultBuilder.append(key+"\n");
			}
			resultBuilder.append("\n");
		}
		appLogger.info("执行结果："+resultBuilder.toString());
		return resultBuilder.toString();
	}

	@Override
	public void close() throws IOException {
		if(adminClient!=null) {
			adminClient.close();
		}
	}
}
